//package com.canal.mq.consumer;
//
//import com.canal.mq.config.ConsumerConfig;
//import com.canal.mq.config.MqConfig;
//import com.canal.mq.listener.CanalListener;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.acl.common.AclClientRPCHook;
//import org.apache.rocketmq.acl.common.SessionCredentials;
//import org.apache.rocketmq.client.AccessChannel;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.apache.rocketmq.remoting.RPCHook;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//
//@Configuration
//@Slf4j
//public class ConsumerClient {
//
//    @Autowired
//    private MqConfig mqConfig;
//    @Autowired
//    private ConsumerConfig consumerConfig;
//    @Value("${spring.profiles.active}")
//    private String active;
//    @Autowired
//    private CanalListener listener;
//
//
//    // 设置为您在阿里云消息队列RocketMQ版控制台上创建的GID,以及替换为您的AccessKey ID和AccessKey Secret。
//    private RPCHook getAclRPCHook() {
//        return new AclClientRPCHook(new SessionCredentials(mqConfig.getAccessKey(), mqConfig.getSecretKey()));
//    }
//
//    @Bean(initMethod = "start", destroyMethod = "shutdown")
//    public DefaultMQPushConsumer buildConsumer() throws Exception {
//
//        /**
//         *创建Producer，并开启消息轨迹。
//         *如果不想开启消息轨迹，可以按照如下方式创建：
//         *DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook());
//         */
//        DefaultMQPushConsumer consumer;
//        if ("pro".equals(active)) {
//            // 生产环境使用阿里云mq
//            consumer = new DefaultMQPushConsumer(consumerConfig.getGroupId(), getAclRPCHook(), new AllocateMessageQueueAveragely());
//        } else {
//            consumer = new DefaultMQPushConsumer(consumerConfig.getGroupId());
//        }
//        // 设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项.
//        consumer.setAccessChannel(AccessChannel.CLOUD);
//        /**
//         *设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。
//         */
//        consumer.setAccessChannel(AccessChannel.CLOUD);
//        /**
//         *设置为您从阿里云消息队列RocketMQ版控制台获取的接入点信息，类似“http://MQ_INST_XXXX.aliyuncs.com:80”。
//         */
//        consumer.setNamesrvAddr(mqConfig.getNameServer());
//        /**
//         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
//         * 如果非第一次启动，那么按照上次消费的位置继续消费
//         */
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        /**
//         * 设置消费模型，集群还是广播，默认为集群
//         */
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        /**
//         * 设置一次消费消息的条数，默认为1条
//         */
//        consumer.setConsumeMessageBatchMaxSize(consumerConfig.getConsumeMessageBatchMaxSize());
//        /**
//         * 注册监听器
//         */
//        consumer.registerMessageListener(listener);
//        try {
//            /**
//             * 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*；如果需要指定订阅该主题下的某些tag，则使用||分割，例如tag1||tag2||tag3
//             */
//            String[] topicTagsArr = consumerConfig.getTopic().split(";");
//            for (String topicTags : topicTagsArr) {
//                String[] topicTag = topicTags.split("~");
//                consumer.subscribe(topicTag[0], topicTag[1]);
//            }
//        } catch (MQClientException e) {
//            log.error("消费者启动失败");
//            throw new Exception(e);
//        }
//        log.info("消费者启动成功，启动环境 = {}，groupId = {}，topic = {}，serverAddr = {}",
//                active, consumerConfig.getGroupId(), consumerConfig.getTopic(), mqConfig.getNameServer());
//        return consumer;
//    }
//
//}